package fox.framework.rpc.endpoint.netty;

import fox.framework.rpc.client.node.RpcNode;
import fox.framework.rpc.common.StringUtils;
import fox.framework.rpc.endpoint.AbstractRpcEndPoint;
import fox.framework.rpc.endpoint.channel.RpcChannel;
import fox.framework.rpc.endpoint.channel.RpcStatusChannel;
import fox.framework.rpc.endpoint.channel.netty.RpcNettyChannel;
import fox.framework.rpc.endpoint.exchange.QueuedSynchronizeExchange;
import fox.framework.rpc.endpoint.exchange.RpcExchange;
import fox.framework.rpc.endpoint.netty.client.FoxClientRequestEncoder;
import fox.framework.rpc.endpoint.netty.client.FoxClientResponseDecoder;
import fox.framework.rpc.endpoint.netty.client.RpcClientEventHandler;
import fox.framework.rpc.endpoint.netty.client.RpcClientResponseHandler;
import fox.framework.rpc.endpoint.netty.heartbeat.RpcHeartBeatDecode;
import fox.framework.rpc.endpoint.netty.heartbeat.RpcHeartBeatEncode;
import fox.framework.rpc.endpoint.netty.server.FoxServerRequestDecoder;
import fox.framework.rpc.endpoint.netty.server.FoxServerResponseEncode;
import fox.framework.rpc.endpoint.netty.server.RpcServerDispatcherHandler;
import fox.framework.rpc.exception.RpcRuntimeException;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.AttributeKey;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


/**
 * @author cuichao
 * @Description: NettyEndpoint
 * @Date: create in 2020/12/29 15:48
 */
public class NettyEndpoint extends AbstractRpcEndPoint {

    public static final AttributeKey<RpcChannel> RPC_CHANNEL_ATTR_KEY = AttributeKey.valueOf("rpc_channel_");

    public static final AttributeKey<Integer> RPC_PING_COUNTER = AttributeKey.valueOf("pingCounter");

    private NettyClient client = new NettyClient();

    private NettyServer server = new NettyServer();

    private RpcExchange exchange = new QueuedSynchronizeExchange();

    private CountDownLatch latch = new CountDownLatch(1);

    private final ConcurrentHashMap<RpcNode, Lock> nodeLock = new ConcurrentHashMap<>();

    private int allIdleTimeSeconds = 0;

    private int readerIdleTimeSeconds = 0;

    private int writerIdleTimeSeconds = 3;

    private int heartbeatMaxTries = 3;


    public NettyEndpoint(String address, int port) {
        this.address = address;
        this.port = port;
    }

    @Override
    public void run() {
        //初始化客户端
        client.init();
        //启动服务
        if (enableServer) {
            server.runServer();
        }

    }

    @Override
    public RpcChannel connection(RpcNode node) throws RpcRuntimeException {
        Lock nodeLock = getNodeLock(node);
        nodeLock.lock();
        try {
            RpcChannel channel = node.getChannel();
            //create
            if (channel == null) {
                Channel nettyChannel = client.connection(node.getHost(), node.getPort());
                channel = new RpcNettyChannel(node, nettyChannel);
                ((RpcChannel.Sync) channel).exchange(exchange);
                nettyChannel.attr(RPC_CHANNEL_ATTR_KEY).set(channel);
                return channel;
            }
            //reConnection
            if (channel instanceof RpcStatusChannel && ((RpcStatusChannel) channel).isAlive()) {
                return channel;
            } else {
                Channel nettyChannel = client.connection(node.getHost(), node.getPort());
                nettyChannel.attr(RPC_CHANNEL_ATTR_KEY).set(node.getChannel());
                ((RpcNettyChannel) channel).setChannel(nettyChannel);
                return channel;
            }
        } finally {
            nodeLock.unlock();
        }
    }


    @Override
    public void shutdown() {
        server.shutdown();
    }


    @Override
    protected void beforeStarted() {
        try {
            //blocking mainThread
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private Lock getNodeLock(RpcNode node) {
        Lock lock = nodeLock.get(node);
        if (lock == null) {
            Lock newLock = new ReentrantLock();
            lock = nodeLock.putIfAbsent(node, newLock);
            return lock == null ? newLock : lock;
        }
        return lock;
    }


    /**
     * netty客户端
     */
    private final class NettyClient {

        private Bootstrap bootstrap = new Bootstrap();
        EventLoopGroup executors = new NioEventLoopGroup(16);

        private void init() {

            RpcClientResponseHandler handler = new RpcClientResponseHandler(exchange);

            bootstrap
                    .group(executors)
                    .channel(NioSocketChannel.class)
                    .attr(RPC_PING_COUNTER, 0)
                    .handler(new LoggingHandler(LogLevel.DEBUG))
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            ch.pipeline().addLast("heartBeatDecoder",new RpcHeartBeatDecode());
                            ch.pipeline().addLast("heartBeatEncoder",new RpcHeartBeatEncode());
                            ch.pipeline().addLast("foxEncoder", new FoxClientResponseDecoder());
                            ch.pipeline().addLast("foxDecoder", new FoxClientRequestEncoder());
                          /*  ch.pipeline().addLast("heartbeatTrigger",new IdleStateHandler(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds));
                            ch.pipeline().addLast("heartbeatHandler", new RpcHeartbeatHandler(new NettyHeartbeat(heartbeatMaxTries)));*/
                            ch.pipeline().addLast("eventPublisher", new RpcClientEventHandler(getPublisher()));
                            ch.pipeline().addLast("clientHandler", handler);
                        }
                    });
        }


        private Channel connection(String address, int port) throws RpcRuntimeException {
            //同步业务线程阻塞
            try {
                ChannelFuture sync = bootstrap.connect(address, port).sync();
                Channel channel = sync.channel();
                return channel;
            } catch (InterruptedException e) {
                throw new RpcRuntimeException(e.getCause());
            }
        }
    }

    /**
     * netty 服务端
     */
    private final class NettyServer {

        private EventLoopGroup master = new NioEventLoopGroup(1);
        private EventLoopGroup worker = new NioEventLoopGroup();
        private ServerBootstrap bootstrap = new ServerBootstrap();


        private void runServer() {

            RpcServerDispatcherHandler handler = new RpcServerDispatcherHandler(NettyEndpoint.this.getDispatcher());

            //添加消息处理管道
            ChannelInitializer<NioSocketChannel> initializer = new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel ch) throws Exception {
                  /*  ch.pipeline().addLast("heartBeatDecoder",new RpcHeartBeatDecode());
                    ch.pipeline().addLast("heartBeatEncoder",new RpcHeartBeatEncode());*/
                    ch.pipeline().addLast("foxDecoder", new FoxServerRequestDecoder());
                    ch.pipeline().addLast("foxEncoder", new FoxServerResponseEncode());
                   /* ch.pipeline().addLast("heartbeatTrigger",new IdleStateHandler(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds));
                    ch.pipeline().addLast("heartbeatHandler", new RpcHeartbeatHandler(new NettyHeartbeat(heartbeatMaxTries)));*/
                    ch.pipeline().addLast("serverHandler", handler);
                }
            };
            bootstrap
                    .group(master, worker)
                    .channel(NioServerSocketChannel.class)
                    .childAttr(RPC_PING_COUNTER, 0)
                    .childHandler(initializer)
                    .handler(new LoggingHandler(LogLevel.DEBUG));
            ChannelFuture channel;
            //绑定IP/Port
            if (StringUtils.isEmpty(address)) {
                channel = bootstrap.bind(port);
            } else {
                channel = bootstrap.bind(address, port);
            }
            // add started listener
            channel.addListeners(future -> {
                if (!future.isSuccess()) {
                    cause = future.cause();
                }
                latch.countDown();
            });

            try {
                //blocking
                channel.sync();
                //blocking
                channel.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                master.shutdownGracefully();
                worker.shutdownGracefully();
            }

        }


        private void shutdown() {
            master.shutdownGracefully();
            worker.shutdownGracefully();
        }


    }


}
